#include "threadpool.h"

const int NUMBER = 2;

threadPool* createThreadPool(const int min, const int max, const int quesSize)
{
    //init 
    threadPool* pool = (threadPool*) malloc(sizeof(threadPool));
    do
    {
        if(pool == NULL)
        {
            printf("malloc threadpool error\n");
            break;
        }
        pool->threadIDs = (pthread_t*) malloc(sizeof(pthread_t) * max);
        if(pool->threadIDs == NULL)
        {
            printf("iids malloc error\n");
            break;
        }
        memset(pool->threadIDs, 0, sizeof(pthread_t) * max);

        pool->minNum = min;
        pool->maxNum = max;
        pool->liveNum = min;
        pool->busyNum = 0;
        pool->exitNum = 0;

        if(pthread_mutex_init(&pool->mutePool, NULL) != 0 ||
           pthread_mutex_init(&pool->muteBusy, NULL) != 0 ||
           pthread_cond_init(&pool->noFull, NULL) != 0 ||
           pthread_cond_init(&pool->noEmpty, NULL) != 0 )
        {
            printf("mutex or cond init error\n ");
            break;
        }
        //任务队列
        pool->taskQ = (Task*) malloc(sizeof(Task) *quesSize);
        pool->queCapcity = quesSize;
        pool->queSize = 0;
        pool->queFront = 0;
        pool->queEnd = 0;

        pool->showDown = 0;
        

        //创建线程
        //管理者线程
        pthread_create(&pool->managerID, NULL, manager, pool);
        //创建工作线程
        for (size_t i = 0; i < min; i++)
        {
            pthread_create(&pool->threadIDs[i], NULL, worker, pool);
        }

        return pool;
    } while (0);

    //释放资源
    if(pool->threadIDs == NULL)
        free(pool->threadIDs);
    if(pool->taskQ == NULL)
        free(pool->taskQ);
    if(pool == NULL)
        free(pool);
    return NULL;  
}

void threadExit(threadPool* pool)
{
    pthread_t pid = pthread_self();
    for (size_t i = 0; i < pool->maxNum; i++)
    {
        if(pool->threadIDs[i] == pid)
        {
            pool->threadIDs[i] = 0;
            printf("threadexit() call, %ld exiting\n", pid);
            break;
        }
    }
    pthread_exit(NULL);    
}

void addTaskTOPool(threadPool* pool, void(*func)(void*), void* arg)
{
    pthread_mutex_lock(&pool->mutePool);
    while (pool->queSize == pool->queCapcity && !pool->showDown)
    {
        //阻塞生产者线程
        pthread_cond_wait(&pool->noFull, &pool->mutePool);
    }
    if(pool->showDown)
    {
        pthread_mutex_unlock(&pool->mutePool);
        return;
    }
    //添加任务
    pool->taskQ[pool->queEnd].function = func;
    pool->taskQ[pool->queEnd].arg = arg;
    pool->queEnd = (pool->queEnd + 1) %pool->queCapcity;
    pool->queSize++;
    
    
    //唤醒阻塞在工作中的线程
    pthread_cond_signal(&pool->noEmpty);

    pthread_mutex_unlock(&pool->mutePool);
}

void* manager(void* arg)
{
    threadPool* pool = (threadPool*)arg;

    while (!pool->showDown)
    {
        //每间隔3秒检测一次
        sleep(3);

        //取出当前线程池中任务数量和当前线程数量
        pthread_mutex_lock(&pool->mutePool);
        int queSize = pool->queSize;
        int liveNum = pool->liveNum;
        pthread_mutex_unlock(&pool->mutePool);

        //取出忙的线程个数
        pthread_mutex_lock(&pool->muteBusy);
        int busyNUm = pool->busyNum;
        pthread_mutex_unlock(&pool->muteBusy);

        //添加线程
        //任务个数 > 存活个数 && 存活个数 < 最大线程
        if(queSize > liveNum && liveNum < pool->maxNum)
        {
            pthread_mutex_lock(&pool->mutePool);
            int counter = 0;
            for (size_t i = 0; 
                 i < pool->maxNum && counter < NUMBER && pool->liveNum < pool->maxNum;
                 i++)
            {
                if(pool->threadIDs[i] == 0)
                {
                    pthread_create(&pool->threadIDs[i], NULL, worker, pool);
                    counter++;
                    pool->liveNum++;
                }
            }       
            pthread_mutex_unlock(&pool->mutePool);
        }
        //销毁线程
        //忙的线程*2  < 存活的线程数 && 存活的线程 > 最小的线程数
        if(busyNUm* 2 < liveNum && liveNum > pool->minNum)
        {
            pthread_mutex_lock(&pool->mutePool);
            pool->exitNum = NUMBER;
            pthread_mutex_unlock(&pool->mutePool);

            //让线程自杀
            for (size_t i = 0; i < NUMBER; i++)
            {
                pthread_cond_signal(&pool->noEmpty);

            }
            

        }


    }
    
}

void* worker(void* arg)
{
    threadPool* pool = (threadPool*)arg;
    while (1)
    {
        pthread_mutex_lock(&pool->mutePool);

        //当前队列是否为空
        while(pool->queSize == 0 &&  !pool->showDown)
        {
            //阻塞工作线程
            //从队列中拿任务进行处理
            pthread_cond_wait(&pool->noEmpty, &pool->mutePool); 
            if(pool->exitNum > 0)
            {
                pool->exitNum--;

                if(pool->liveNum > pool->minNum)
                    pool->liveNum--;

                pthread_mutex_unlock(&pool->mutePool);
                threadExit(pool);
            }
        }

        //判断当前的线程是否关闭
        if(pool->showDown)
        {
            pthread_mutex_unlock(&pool->mutePool);
            threadExit(pool);
        }

        //从任务队列中取出任务处理
        Task task;
        task.function = pool->taskQ[pool->queFront].function;
        task.arg = pool->taskQ[pool->queFront].arg;

        //移动头结点
        pool->queFront = (pool->queFront+1) % pool->queCapcity;
        pool->queSize--;
        
        pthread_cond_signal(&pool->noFull);

        pthread_mutex_unlock(&pool->mutePool);

        //开始处理任务
        printf("start %ld work number = %d\n", pthread_self(), *(int*)task.arg);
        pthread_mutex_lock(&pool->muteBusy);
        pool->busyNum++;
        task.function(task.arg);
        pthread_mutex_unlock(&pool->muteBusy);

        printf("end %ld work number = %d\n", pthread_self(), *(int*)task.arg);
        pthread_mutex_lock(&pool->muteBusy);
        task.function(task.arg);
        task.arg = NULL;
        pool->busyNum--;
        pthread_mutex_unlock(&pool->muteBusy);
    }
    
}

//销毁线程池
int destoryThreadPool(threadPool* pool)
{
    if(pool == NULL)
        return -1;
    //关闭线程池
    pool->showDown = 1;

    //阻塞回收管理者线程
    pthread_join(pool->managerID,NULL);

    //唤醒阻塞的消费者线程
    for (size_t i = 0; i < pool->liveNum; i++)
    {
        pthread_cond_signal(&pool->noEmpty);
    }

    //释放堆内存
    if(pool->taskQ)
    {
        free(pool->taskQ);
        pool->taskQ = NULL;
    }    
    if(pool->threadIDs)
    {
        free(pool->threadIDs);
        pool->threadIDs = NULL;
    }    

    //释放互斥锁和条件变量
    pthread_cond_destroy(&pool->noFull);
    pthread_cond_destroy(&pool->noEmpty);
    pthread_mutex_destroy(&pool->mutePool);
    pthread_mutex_destroy(&pool->muteBusy);
    
    if(pool)
    {
        free(pool);
        pool = NULL;
    }
        
    
    
    return 0;
}

//获取工作的线程个数
int threadLiveNum(threadPool* pool)
{
    pthread_mutex_lock(&pool->muteBusy);
    int num = pool->busyNum;
    pthread_mutex_unlock(&pool->muteBusy);
    return num;
}

//获取工作的busy的线程个数
int threadBusyNum(threadPool* pool)
{
    pthread_mutex_lock(&pool->mutePool);
    int num = pool->liveNum;
    pthread_mutex_unlock(&pool->mutePool);
    return num;
}
